# coding=utf-8
#从数据库读取数据,生成新数据,新数据包括涨停情况,涨停个数,涨停板数,涨停情况,上下交易日情况.多进程计算,已有数据不会重新计算,以加快计算速度.
from __future__ import print_function, absolute_import
from gevent import spawn, joinall
import pandas as pd
import time
import datetime
import tqdm
from concurrent.futures import *
import chage_data_type
import os
from multiprocessing import Queue, JoinableQueue, Process
import asyncio
max_workers = os.cpu_count()

'''
try:
    import myzmode
    print('成功连接服务器,已登录相关终端...')
except KeyError:
    print('没能成功连接服务器,请确认已登录相关终端!!!!!')
except OSError:
    pass
'''


def last_bss(df):
    bs = df.set_index('symbol').sort_values('trade_date').iloc[-1, :]['bs']
    return bs


def last_data(df):
    bs = df.sort_values('trade_date').iloc[-1, :]
    return bs


def change_format(code):
    code_split = code.split('.', 1)
    return code_split[0]


def get_data():
    def new_data(date_read):
        s = time.time()
        ih = pd.read_hdf('D:/data/history_instruments.h5', 'history_instruments',
                         where="trade_date>" + ' ''"' + date_read + '"').dropna()
        day = pd.read_hdf('D:/data/history_daily.h5', 'history_daily', where="trade_date>" + ' ''"' + date_read + '"').dropna()
        turn = pd.read_hdf('D:/data/turn_daily_gm.h5', 'turn_daily_gm', where="trade_date>" + ' ''"' + date_read + '"').dropna().drop_duplicates()
        end = time.time()
        print('耗时 ' + str(end - s) + ' 秒')
        print('读取完成!开始整合数据...')
        ih = ih[ih['is_suspended'] == 0]
        ih = ih[['trade_date', 'symbol', 'upper_limit', 'sec_level', 'is_suspended']]
        # turn = turn.rename(columns={'date': 'trade_date'})
        all_code = pd.merge(ih, day, how='inner', on=['symbol', 'trade_date'])
        print('下面为bug所在!')
        all_code = pd.merge(all_code, turn, how='inner', on=['symbol', 'trade_date'])
        # all_code = all_code.set_index('symbol')
        # all_code = all_code.drop_duplicates()
        all_code['touch'] = all_code['high'] >= all_code['upper_limit']
        all_code['limit_up'] = all_code['close'] >= all_code['upper_limit']
        all_code['line'] = all_code['low'] >= all_code['upper_limit']
        all_code['o_l'] = all_code['open'] >= all_code['upper_limit']
        all_code = all_code.drop_duplicates().sort_values('trade_date')
        #all_code = all_code[all_code['trade_date'] > date_read].groupby('symbol').apply(last_data)
        print('数据整合完毕')
        print('准备工作完成...')
        return all_code
    if os.path.exists('D:/data/limit_up.h5'):
        print('开始读取现有数据...')
        try:
            last = pd.read_hdf('D:/data/limit_up.h5', 'limit_up', where="trade_date>'2010-04-01'")#, columns=['symbol', 'trade_date', 'bs'])
        except KeyError:
            print('数据库没有指定的表,请考虑删除D:/data/limit_up.h5数据库!')
        def last_data(df):
            df_last = df.iloc[-1]
            return df_last#[['trade_date', 'bs']]
        # 下面用apply选出最后一行
        last = last.groupby('symbol').apply(last_data)
        del last['symbol']
        last = last.reset_index()
        #last = last[['symbol', 'trade_date', 'rush', 'bs', 'limit_up', 'touch']]
        date_read = last['trade_date'].dt.strftime('%Y-%m-%d').max()
        old_data = new_data(date_read)
        if not old_data.empty:
            new = pd.merge(old_data, last, how='outer').sort_values('trade_date')
            print('已存在相关数据,开始追加模式')
            return new#, last
    else:
        store = pd.HDFStore('D:/data/limit_up.h5')
        store.close()
        new = new_data('2010-04-10')
        '''
        def new_data(df):
            df_last = df.iloc[0]
            return df_last#[['trade_date', 'bs']]
        # 下面用apply选出最后一行
        last = new.groupby('symbol').apply(new_data)
        del last['symbol']
        last = last.reset_index()
        '''
        print('没有相关数据,开始覆盖模式')
        return new#, last

def app_data(l_split):

    def split_data(df):
        last = df[1].iloc[0]
        try:
            m, n = last['rush'], last['bs']
            if m > 0:
                m = m-1
            if n > 0:
                n = n-1
        except KeyError:
            m, n = 0, 0
        # 当日有无封板
        def bs(x):
            global n
            if x == False:
                n = 0
            else:
                try:
                    n += 1
                except NameError:
                    n = 0
            return n

        # 当日有无冲板
        def rush(x):
            global m
            if x == False:
                m = 0
            else:
                try:
                    m += 1
                except NameError:
                    m = 0
            return m

        all_code = df[1]
        #all_code = df[1].dropna(axis=1)
        try:
            data = all_code.sort_values('trade_date')
            data = data.set_index('trade_date')
        except:
            print(all_code.head())
        # data['symbol'] = code
        data['bs'] = data['limit_up'].apply(bs)
        data['rush'] = data['touch'].apply(rush)
        data['o_pct'] = data['open'] / data['pre_close'] - 1
        data['h_pct'] = data['high'] / data['pre_close'] - 1
        data['l_pct'] = data['low'] / data['pre_close'] - 1
        data['c_pct'] = data['close'] / data['pre_close'] - 1
        # data['max'] = data.groupby('trade_date')['bs'].transform('max')
        data = data.reset_index().sort_values('trade_date')
        yesterday = data.groupby('symbol').shift(1)
        yesterday = yesterday[
            ['bs', 'TURNRATE', 'o_pct', 'o_l', 'h_pct', 'l_pct', 'line', 'c_pct', 'vol']]
        yesterday = yesterday.rename(columns={'bs': 'ybs', 'o_pct': 'yo_pct', 'TURNRATE': 'y_turn', 'vol': 'ytd_vol',
                                              'c_pct': 'yc_pct', 'o_l': 'y_o_l', 'h_pct': 'yh_pct',
                                              'l_pct': 'yl_pct', 'line': 'y_line'})
        data[['ybs', 'y_turn', 'yo_pct', 'y_o_l', 'yh_pct', 'yl_pct', 'y_line', 'yc_pct', 'ytd_vol']] = yesterday
        '''
        try:
            data.loc[0] = last[['ybs', 'yo_pct', 'y_turn', 'ytd_vol', 'yc_pct', 'y_o_l', 'yh_pct', 'yl_pct', 'y_line', 'ytd']]
        except KeyError:
            pass
        '''
        tomorrow = data.groupby('symbol').shift(-1)
        tomorrow = tomorrow[['o_pct', 'h_pct', 'l_pct', 'c_pct']]
        tomorrow = tomorrow.rename(columns={'h_pct': 'nh_pct',
                                            'o_pct': 'no_pct', 'c_pct': 'nc_pct', 'l_pct': 'nl_pct'})
        data[['no_pct', 'nh_pct', 'nl_pct', 'nc_pct']] = tomorrow
        #data[['n_turnover', 'nh_pct', 'no_pct', 'nc_pct', 'nl_pct', 'n_bs']].astype('float64')
        #data.loc[len(data)-1] = last[['n_turnover', 'nh_pct', 'tmw', 'no_pct', 'nc_pct', 'nl_pct', 'n_bs']]
        #data = pd.merge(yesterday, data, how='left')
        #data = pd.merge(tomorrow, data, how='left')
        #data = pd.concat([data, yesterday, tomorrow], axis=1, join='outer')
        data = data.shift(-1).drop(data.index[-1])
        #data[['limit_up', 'line', 'o_l', 'touch', 'y_line', 'y_o_l']] = data[['limit_up', 'line', 'o_l', 'touch',
                                                                              #'y_line', 'y_o_l']].copy().astype('bool')
        data[['ybs', 'sec_level', 'rush', 'bs']] = data[['ybs', 'sec_level', 'rush', 'bs'
                                                         ]].copy().fillna('0').astype('uint8')
        '''
        data = data[['symbol', 'trade_date', 'rush', 'bs', 'o_pct', 'h_pct', 'l_pct', 'c_pct', 'touch', 'line',
                     'amount', 'o_l', 'ybs', 'sec_level', 'y_o_l', 'y_line', 'no_pct', 'nl_pct', 'nh_pct', 'nc_pct',
                     'yo_pct', 'yc_pct', 'yh_pct', 'yl_pct', 'TURNRATE', 'y_turn', 'is_suspended', 'upper_limit',
                     'ytd_vol', 'vol', 'limit_up']]
        '''
        if df[0] == 'SHSE.600070':
            print(data[['symbol', 'trade_date', 'TURNRATE']].tail(30))
        return data
        '''
        try:
            data.to_hdf('D:/data/limit_up.h5', 'limit_up', append=True, format='t', index=False,
                        data_columns=['symbol', 'trade_date', 'bs'])
        except TypeError:
            pass
        except:
            print('发生未知错误!')
    all_task = [spawn(split_data, df) for df in l_split]
    joinall(all_task)
    results = [task.result() for task in all_task]
    print(results)
    return results
            '''
    c = [split_data(df) for df in l_split]
    results = pd.concat(c)
    return results

def mp(l_split):

    def split_data(df):
        last = df[1].iloc[0]
        try:
            m, n = last['rush'], last['bs']
            if m > 0:
                m = m-1
            if n > 0:
                n = n-1
        except KeyError:
            m, n = 0, 0
        # 当日有无封板
        def bs(x):
            global n
            if x == False:
                n = 0
            else:
                n += 1
            return n

        # 当日有无冲板
        def rush(x):
            global m
            if x == False:
                m = 0
            else:
                m += 1
            return m

        all_code = df[1]
        #all_code = df[1].dropna(axis=1)
        try:
            data = all_code.sort_values('trade_date')
            data = data.set_index('trade_date')
        except:
            print(all_code.head())
        # data['symbol'] = code
        data['bs'] = data['limit_up'].apply(bs)
        data['rush'] = data['touch'].apply(rush)
        data['o_pct'] = data['open'] / data['pre_close'] - 1
        data['h_pct'] = data['high'] / data['pre_close'] - 1
        data['l_pct'] = data['low'] / data['pre_close'] - 1
        data['c_pct'] = data['close'] / data['pre_close'] - 1
        # data['max'] = data.groupby('trade_date')['bs'].transform('max')
        data = data.reset_index().sort_values('trade_date')
        yesterday = data.groupby('symbol').shift(1)
        yesterday = yesterday[
            ['bs', 'TURNRATE', 'o_pct', 'o_l', 'h_pct', 'l_pct', 'line', 'trade_date', 'c_pct', 'vol']]
        yesterday = yesterday.rename(columns={'bs': 'ybs', 'o_pct': 'yo_pct', 'TURNRATE': 'y_turn', 'vol': 'ytd_vol',
                                              'c_pct': 'yc_pct', 'o_l': 'y_o_l', 'h_pct': 'yh_pct',
                                              'l_pct': 'yl_pct', 'line': 'y_line', 'trade_date': 'ytd'})
        tomorrow = data.groupby('symbol').shift(-1)
        tomorrow = tomorrow[['bs', 'TURNRATE', 'o_pct', 'h_pct', 'l_pct', 'trade_date', 'c_pct']]
        tomorrow = tomorrow.rename(columns={'TURNRATE': 'n_turnover', 'h_pct': 'nh_pct', 'trade_date': 'tmw',
                                            'o_pct': 'no_pct', 'c_pct': 'nc_pct', 'l_pct': 'nl_pct', 'bs': 'n_bs'})
        #data = pd.merge(yesterday, data, how='left')
        #data = pd.merge(tomorrow, data, how='left')
        data = pd.concat([data, yesterday, tomorrow], axis=1, join='outer')
        data = data.shift(-1).drop(data.index[-1])
        #data[['limit_up', 'line', 'o_l', 'touch', 'y_line', 'y_o_l']] = data[['limit_up', 'line', 'o_l', 'touch',
                                                                              #'y_line', 'y_o_l']].copy().astype('bool')
        data[['ybs', 'sec_level', 'rush', 'bs']] = data[['ybs', 'sec_level', 'rush', 'bs'
                                                         ]].copy().fillna('0').astype('uint8')
        data = data[['symbol', 'trade_date', 'rush', 'bs', 'o_pct', 'h_pct', 'l_pct', 'c_pct', 'touch', 'line',
                     'amount', 'o_l', 'ybs', 'sec_level', 'y_o_l', 'y_line', 'no_pct', 'nl_pct', 'nh_pct', 'nc_pct',
                     'yo_pct', 'yc_pct', 'yh_pct', 'yl_pct', 'TURNRATE', 'y_turn', 'is_suspended', 'upper_limit',
                     'ytd_vol', 'vol', 'limit_up', 'pre_close']]
        return data
        '''
        try:
            data.to_hdf('D:/data/limit_up.h5', 'limit_up', append=True, format='t', index=False,
                        data_columns=['symbol', 'trade_date', 'bs'])
        except TypeError:
            pass
        except:
            print('发生未知错误!')
    all_task = [spawn(split_data, df) for df in l_split]
    joinall(all_task)
    results = [task.result() for task in all_task]
    print(results)
    return results
            '''
    c = [split_data(df) for df in l_split]
    results = pd.concat(c)
    return results

def mt_por():
    l_result = []
    #store = pd.HDFStore('D:/data/limit_up.h5')
    #store.close()
    try:
        l = get_data()
    except AttributeError:
        print('表格没有数据!')
    try:
        l_data = list(l.groupby('symbol'))
        l_code = [l_data[n_split:n_split + 100] for n_split in range(0, len(l_data), 100)]
        with ProcessPoolExecutor(max_workers) as ppe:
            all_task = [ppe.submit(app_data, data_split) for data_split in l_code]
            with tqdm.tqdm(total=len(l_code)) as tq:
                for future in as_completed(all_task):
                    data = future.result()
                    l_result.append(data)
                    tq.update()
        data = pd.concat(l_result)
        data['ytd_max'] = data.groupby('trade_date')['ybs'].transform('max')
        data['max'] = data.groupby('trade_date')['bs'].transform('max')
        try:
            del data['count']
        except KeyError:
            pass
        df = data[data['ybs'] > 0].groupby(['trade_date'])['ybs'].agg('count')
        df = df.reset_index()
        df = df.rename(columns={'ybs': 'count'})
        data = pd.merge(data, df, how='left', on='trade_date')
        print('追加数据因dtype不一致,所以dropna,可能导致数据异常!')
        data = data.fillna(0)
        data[['ytd_max', 'max']] = data[['ytd_max', 'max']].astype('uint8')
        data['is_suspended'] = data['is_suspended'].astype('uint8')
        data['count'] = data['count'].astype('uint32')
        data[['vol', 'amount']] = data[['vol', 'amount']].astype('uint64')
        data.to_hdf('D:/data/limit_up.h5', 'limit_up', append=True, format='t', index=False,
                    data_columns=['symbol', 'trade_date', 'bs'])
    except AttributeError:
        print('没有数据需要更新!')


if __name__ == '__main__':
    mt_por()
